CSVファイルを使ったCognitoのバックアップとリストア
2023.12.12はじめに
私の担当しているプロジェクトではユーザー認証にCognitoユーザープールを使っています。Cognitoユーザープールのバックアップを作成する方法を調べました。
調べてみたところ、Cognito User Profiles Export リファレンスアーキテクチャ というものがありました。よさそうだと思ったのですが、ユーザープールで多要素認証が有効になっている場合は使えない、という制約があり、私の担当しているプロジェクトでは利用することができませんでした。
代替案としてCSVファイルを使ったバックアップとリストアができるかを確認したのですが問題なくできたので方法をまとめます。
バックアップの作成
CSVファイルをS3に保存するLambda Functionが以下になります。言語はPython3.10です。このLambda FunctionをEventBridgeを使って定期的に実行します。
Cognitoの設定によって、バックアップのために必要な情報が変わると思います。私のプロジェクトではメールアドレス、もしくはユーザー名をログインIDにしています。電話番号は使っていません。
50行目以降のループ内でCognitoから取得したユーザー情報を1件ずつオブジェクトに詰めている部分がありますので、そちらを環境に合わせて書き換えてください。
CSVファイルは指定したS3のバケットにアップロードされます。CSVファイルにはメールアドレスや電話番号など顧客の情報が入っているのでバケットポリシーや暗号化などセキュリティに関する設定をしておきましょう。
import json
import boto3
import csv
from botocore.exceptions import ClientError
def paginate_cognito_users(user_pool_id, page_size=60):
"""
Cognitoユーザープールからユーザーデータをページネーションして取得するジェネレーター関数
"""
cognito_client = boto3.client('cognito-idp')
pagination_token = None
while True:
try:
if pagination_token:
response = cognito_client.list_users(
UserPoolId=user_pool_id,
PaginationToken=pagination_token,
Limit=page_size
)
else:
response = cognito_client.list_users(
UserPoolId=user_pool_id,
Limit=page_size
)
yield response['Users']
pagination_token = response.get('PaginationToken')
if not pagination_token:
break
except ClientError as e:
print(f"ユーザーの一覧取得エラー: {e}")
raise
def lambda_handler(event, context):
# Cognitoのユーザープール設定
user_pool_id = "your_user_pool_id"
# S3設定
s3_bucket = "your_s3_bucket_name"
s3_key = "backup.csv"
# 一時ファイルの保存先
temp_file_path = '/tmp/backup.csv'
# CSVファイルに書き込むユーザーデータ
csv_data = []
# ページネーションを考慮してCognitoからユーザーデータを取得
for users in paginate_cognito_users(user_pool_id):
for user in users:
# 必要なユーザー属性を含む辞書を作成
user_attributes = {}
user_attributes["email"] = next((attr['Value'] for attr in user['Attributes'] if attr['Name'] == "email"), "")
user_attributes["email_verified"] = next((attr['Value'] for attr in user['Attributes'] if attr['Name'] == "email_verified"), "")
user_attributes["phone_number_verified"] = False
user_attributes["cognito:mfa_enabled"] = False
user_attributes["cognito:username"] = user['Username']
csv_data.append(user_attributes)
# CSVにデータを書き込み
with open(temp_file_path, 'w', newline='') as csv_file:
# Cognitoユーザープールに必要なフィールド名
cognito_fieldnames = [
"name", "given_name", "family_name", "middle_name", "nickname",
"preferred_username", "profile", "picture", "website", "email",
"email_verified", "gender", "birthdate", "zoneinfo", "locale",
"phone_number", "phone_number_verified", "address", "updated_at",
"cognito:mfa_enabled", "cognito:username"
]
writer = csv.DictWriter(csv_file, fieldnames=cognito_fieldnames)
writer.writeheader()
for row in csv_data:
writer.writerow(row)
# S3にCSVをアップロード
s3_client = boto3.client('s3')
try:
s3_client.upload_file(temp_file_path, s3_bucket, s3_key)
except ClientError as e:
print(f"S3へのCSVアップロードエラー: {e}")
return {
'statusCode': 500,
'body': json.dumps('CSV upload error to S3.')
}
return {
'statusCode': 200,
'body': json.dumps('CSV backup was successful.')
}
LambdaにアタッチするIAMポリシーは以下です。Resourceはすべてのリソースを対象にしているので、必要に応じてResourceを書き換えてください。
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"cognito-idp:ListUsers"
],
"Resource": "*"
},
{
"Effect": "Allow",
"Action": [
"s3:PutObject"
],
"Resource": "*"
}
]
}
リストアする
Management Consoleから新しいユーザープールを作成後、作成したCSVファイルから新しいユーザープールにデータをリストアします。CognitoにはCSVからデータをインポートする機能がありますのでそれを使います。Management Consoleでユーザープールの画面を開くと、一番下にユーザーをインポートという項目があります。「インポートジョブを作成」ボタンを押してください。
「インポートジョブを作成」画面ではジョブ名、作成するロール名、CSVを指定するだけです。ジョブ名、作成するロール名は何でもいいので分かりやすい名前をつけてください。右下のあたりにtemplate.csv というリンクがあり、これを見ると必要なCSVのヘッダーを確認できます。もしインポート時にエラーが出た場合はヘッダーに違いがないかを確認してください。
「ジョブを作成および開始」ボタンを押すとインポートが始まります。 450件ほどですが、数秒でリストアできました。
パスワードはバックアップされないのでパスワードリセットが必要です。私の担当しているプロジェクトではMFAはオプションでした。ユーザープールが変わりMFAは使えないので、一旦すべて無効にしています（Lambda Functionの57行目参照）。そのためログイン後に再設定が必要です。 パスワードリセット後にログインするところまで確認できました。
調べたことは以上になります。最後までご覧いただきありがとうございました！
